package com.shujia.flink.tf

import org.apache.flink.api.java.functions.KeySelector
import org.apache.flink.streaming.api.scala._

object Demo4KeyBY {
  def main(args: Array[String]): Unit = {

    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment

    val linesDS: DataStream[String] = env.socketTextStream("master", 8888)

    val kvDS: DataStream[(String, Int)] = linesDS.flatMap(_.split(",")).map((_, 1))


    /**
      * keyBy： 将相同的key 发送到同一个task中， key之后可以进行聚合计算，针对每一个key
      *
      */
    //scala api
    val scalaKeyByDS: KeyedStream[(String, Int), String] = kvDS.keyBy(_._1)


    //java api
    val javaKeyByDS: KeyedStream[(String, Int), String] = kvDS.keyBy(new KeySelector[(String, Int), String] {
      /**
        * 获取key,每一条数据执行一次
        *
        * @param value : 一行数据
        * @return
        */
      override def getKey(value: (String, Int)): String = {
        value._1
      }
    })

    javaKeyByDS.print()


    env.execute()

  }

}
